#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
p21_user_portrait.py
核心优化：1.修复字段映射颠倒 2.修正完播计算逻辑 3.匹配标签分隔符 4.设置category为“陪玩”
解决 Illegal mix of collations 错误，确保与viewlogs、user_portraits表结构一致
"""
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text, inspect
from sqlalchemy import MetaData, Table, Column
from sqlalchemy.types import Text, BigInteger, Float
import pymysql
import urllib.parse  # 新增：用于URL编码

# 数据库连接配置
db_config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': '123456',
    'database': 'mydb'
}

# 工具函数1：SQL字符串转义
def escape_sql_string(val):
    if pd.isna(val):
        return 'NULL'
    str_val = str(val).replace("'", "''")
    return f"'{str_val}'"

# 工具函数2：校验必需表（viewlogs + video_label）及字段
def check_required_tables(engine):
    inspector = inspect(engine)
    # 1. 校验表是否存在
    all_tables = inspector.get_table_names()
    required_tables = ['viewlogs', 'video_label']
    missing_tables = [t for t in required_tables if t not in all_tables]
    if missing_tables:
        raise ValueError(f"缺少必需表：{missing_tables}（需先创建video_label表）")
    
    # 2. 校验viewlogs表字段（匹配图1：user_id、play_duration、user_like、user_comment、user_gift、video_id等）
    viewlogs_cols = [col['name'] for col in inspector.get_columns('viewlogs')]
    viewlogs_required = ['user_id', 'play_duration', 'user_like', 'user_comment', 'user_gift', 'video_id', 'total_likes']
    viewlogs_missing = [f for f in viewlogs_required if f not in viewlogs_cols]
    if viewlogs_missing:
        raise ValueError(f"viewlogs表缺少必需字段：{viewlogs_missing}（需包含video_id用于关联标签）")
    
    # 3. 校验video_label表字段（video_id、label_name）
    video_label_cols = [col['name'] for col in inspector.get_columns('video_label')]
    video_label_required = ['video_id', 'label_name']
    video_label_missing = [f for f in video_label_required if f not in video_label_cols]
    if video_label_missing:
        raise ValueError(f"video_label表缺少必需字段：{video_label_missing}（需包含video_id和label_name）")
    
    print(f"\n 表校验通过：viewlogs和video_label表均存在")
    print(f"viewlogs表字段：{viewlogs_cols}")
    print(f"video_label表字段：{video_label_cols}")
    return viewlogs_cols, video_label_cols

try:
    # 1. 数据库连接（对密码进行URL编码处理特殊字符）
    encoded_password = urllib.parse.quote_plus(db_config['password'])  # 编码密码中的特殊字符
    engine = create_engine(f"mysql+pymysql://{db_config['user']}:{encoded_password}@{db_config['host']}:{db_config['port']}/{db_config['database']}")
    print("数据库连接成功！")
    
    # 2. 校验必需表和字段（确保与图1 viewlogs表结构匹配）
    viewlogs_cols, video_label_cols = check_required_tables(engine)
    has_video_id = 'video_id' in viewlogs_cols  # 已通过校验，必然为True
    has_duration = 'duration' in viewlogs_cols  # 图1无此字段，固定为False
    has_category = 'category' in viewlogs_cols  # 图1无此字段，固定为False
    
    # 3. 构造多表关联查询SQL（核心修改1：将category设为“陪玩”，而非“未知分类”）
    query = f"""
        SELECT 
            v.user_id, 
            v.play_duration, 
            v.user_like, 
            v.user_comment, 
            v.user_gift, 
            v.total_likes, 
            v.video_id,
            -- 关联video_label获取标签，无匹配时设为“未知标签”
            COALESCE(l.label_name, '未知标签') AS labels,
            -- 核心修改：固定填充“陪玩”，而非“未知分类”
            '陪玩' AS category,
            -- 视频总时长：按play_duration*3计算（图1无duration字段，固定此逻辑）
            v.play_duration * 3 AS duration
        FROM viewlogs v
        -- 左连接：保留所有viewlogs记录，显式指定COLLATE避免排序规则错误
        LEFT JOIN video_label l 
            ON v.video_id COLLATE utf8mb4_general_ci = l.video_id COLLATE utf8mb4_general_ci
    """
    print(f"\n执行多表关联查询SQL：\n{query}")
    df_in = pd.read_sql(query, engine)
    print(f"从viewlogs+video_label读取数据，共 {len(df_in)} 条记录。")
    
    # 4. 空数据判断（创建空表，严格匹配图2 user_portraits结构）
    if len(df_in) == 0:
        metadata = MetaData()
        user_portraits = Table(
            'user_portraits', metadata,
            Column('user_id', Text, comment='用户ID'),
            Column('play_videos', BigInteger, comment='播放视频数量'),
            Column('play_duration', BigInteger, comment='总播放时长'),
            Column('duration', BigInteger, comment='视频总时长'),
            Column('play_rate', Float, comment='播放率'),
            Column('completions', BigInteger, comment='完播数量'),
            Column('likes', BigInteger, comment='喜欢数量'),
            Column('comments', BigInteger, comment='评论数量'),  # 修复：原favorites改为comments
            Column('gifts', BigInteger, comment='打赏数量'),       # 修复：原reposts改为gifts
            Column('category', Text, comment='分类（固定为陪玩）'),  # 核心修改2：注释说明category固定为“陪玩”
            Column('labels', Text, comment='标签')
        )
        metadata.create_all(engine, checkfirst=True)
        print("无数据，已创建空的user_portraits表（category字段默认后续填充为“陪玩”）")
        exit(0)
    
    # 5. 数据预处理（字段映射、类型转换、空值清洗）
    # 字段映射：修复颠倒问题，匹配图2字段含义
    column_mapping = {
        'user_like': 'likes',       # viewlogs.user_like → user_portraits.likes（喜欢数量）
        'user_comment': 'comments', # viewlogs.user_comment → user_portraits.comments（评论数量）
        'user_gift': 'gifts'        # viewlogs.user_gift → user_portraits.gifts（打赏数量）
    }
    df_in = df_in.rename(columns=column_mapping)
    
    # 数值字段类型转换（避免计算错误，匹配MySQL BIGINT类型）
    numeric_fields = ['play_duration', 'duration', 'gifts', 'total_likes', 'likes', 'comments']
    for field in numeric_fields:
        df_in[field] = pd.to_numeric(df_in[field], errors='coerce')  # 无法转换设为NaN
    
    # 空值清洗：移除无效字段（duration为计算字段，无需清洗；保留核心原始字段）
    drop_subset = ['user_id', 'play_duration', 'video_id']
    df_in = df_in.dropna(subset=drop_subset)
    print(f"数据清洗后剩余 {len(df_in)} 条记录。")
    
    # 6. 标签处理（匹配图2逗号分隔格式，拆分多标签）
    df_in["labels"] = df_in["labels"].str.split(',')
    df = df_in.explode("labels").dropna(subset=["labels"])  # 展开多标签
    df["labels"] = df["labels"].str.strip()  # 去除标签前后空格
    df = df[df["labels"] != ""]  # 过滤空字符串标签
    print(f"标签拆分后剩余 {len(df)} 条记录（匹配图2逗号分隔格式）")
    
    # 7. 核心指标计算（修复完播逻辑，匹配实际业务场景）
    # 完播：播放时长≥视频总时长的80%记为1（修复原“≥总时长”的不合理逻辑）
    df["completions"] = (df["play_duration"] >= df["duration"] * 0.8).astype(int)
    # 喜欢：user_like=1记为1，其他为0（匹配图1 user_like取值：0/1/Null）
    df["likes"] = df["likes"].apply(lambda x: 1 if x == 1 and not pd.isna(x) else 0)
    # 评论：有评论内容（非空/非Null）记为1（匹配图1 user_comment字段特性）
    df["comments"] = df["comments"].apply(lambda x: 1 if str(x).strip() != "" and not pd.isna(x) else 0)
    # 打赏：user_gift!=-1且非空记为1（匹配图1 user_gift=-1表示未打赏的规则）
    df["gifts"] = df["gifts"].apply(lambda x: 1 if x != -1 and not pd.isna(x) else 0)
    
    # 8. 按用户+标签+分类聚合（生成用户画像核心数据，category已为“陪玩”）
    group_by_cols = ["user_id", "labels", "category"]
    df_agg = df.groupby(group_by_cols).agg(
        play_videos=("video_id", "nunique"),  # 播放视频数量：去重计数
        play_duration=("play_duration", "sum"),  # 总播放时长：求和
        duration=("duration", "sum"),  # 视频总时长：求和
        play_rate=("play_duration", lambda x: 
            round(x.sum() / df.loc[x.index, "duration"].sum(), 4) if df.loc[x.index, "duration"].sum() != 0 else 0
        ),  # 播放率：保留4位小数
        completions=("completions", "sum"),  # 完播数量：求和
        likes=("likes", "sum"),  # 喜欢数量：求和
        comments=("comments", "sum"),  # 评论数量：求和
        gifts=("gifts", "sum")  # 打赏数量：求和
    ).reset_index()
    
    # 9. 字段最终修正（严格匹配图2 user_portraits字段顺序和类型）
    final_cols = [
        "user_id", "play_videos", "play_duration", "duration", "play_rate",
        "completions", "likes", "comments", "gifts", "category", "labels"
    ]
    df_agg = df_agg[final_cols]
    
    # 数值字段类型转换（匹配MySQL BIGINT/DOUBLE）
    bigint_fields = ['play_videos', 'play_duration', 'duration', 'completions', 'likes', 'comments', 'gifts']
    for field in bigint_fields:
        df_agg[field] = df_agg[field].astype(np.int64)
    df_agg['play_rate'] = df_agg['play_rate'].astype(np.float64)
    
    # 10. 排序（按用户ID升序、播放率降序）
    df_agg = df_agg.sort_values(by=['user_id', 'play_rate'], ascending=[True, False])
    print("数据聚合与排序完成！")
    
    # 11. 写入user_portraits表
    df_agg.to_sql(
        name='user_portraits',
        con=engine,
        if_exists='replace',  # 测试阶段用replace，生产环境建议改为append
        index=False,
        dtype={
            'user_id': Text(),
            'play_videos': BigInteger(),
            'play_duration': BigInteger(),
            'duration': BigInteger(),
            'play_rate': Float(),
            'completions': BigInteger(),
            'likes': BigInteger(),
            'comments': BigInteger(),
            'gifts': BigInteger(),
            'category': Text(),  # 写入的category值为“陪玩”
            'labels': Text()
        }
    )
    print(f"已将 {len(df_agg)} 条用户画像数据写入user_portraits表")
    
    # 12. 生成SQL脚本（备份用，插入的category值为“陪玩”）
    sql_file_path = 'user_portraits.sql'
    with open(sql_file_path, 'w', encoding='utf-8') as f:
        # 创建表SQL（category注释说明为“陪玩”）
        f.write("""
DROP TABLE IF EXISTS `user_portraits`;
CREATE TABLE `user_portraits` (
    `user_id` TEXT COMMENT '用户ID',
    `play_videos` BIGINT COMMENT '播放视频数量',
    `play_duration` BIGINT COMMENT '总播放时长',
    `duration` BIGINT COMMENT '视频总时长',
    `play_rate` DOUBLE COMMENT '播放率',
    `completions` BIGINT COMMENT '完播数量',
    `likes` BIGINT COMMENT '喜欢数量',
    `comments` BIGINT COMMENT '评论数量',
    `gifts` BIGINT COMMENT '打赏数量',
    `category` TEXT COMMENT '分类（固定为陪玩）',
    `labels` TEXT COMMENT '标签'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
""")
        # 批量插入数据（插入的category值为“陪玩”）
        batch_size = 1000
        total_rows = len(df_agg)
        for i in range(0, total_rows, batch_size):
            batch_df = df_agg.iloc[i:i+batch_size]
            values = [f"({','.join([escape_sql_string(val) for val in row])})" for _, row in batch_df.iterrows()]
            f.write(f"INSERT INTO `user_portraits` ({','.join(final_cols)}) VALUES {','.join(values)};\n")
    print(f"\n执行完成！SQL脚本已保存至 {sql_file_path}")

except ValueError as ve:
    print(f"\n配置错误：{ve}")
except Exception as e:
    print(f"\n执行错误：{str(e)}")
    import traceback
    traceback.print_exc()